package edu.nepu.flink.api.checkpoint;

/**
 * @Date 2024/3/3 16:10
 * @Created by chenshuaijun
 */
public class CheckpointOverview {

    public static void main(String[] args) {
        /**
         * TODO checkpoint的概述: 基于的算法是chandy-Lamport(异步分界线快照)
         * checkpoint的概念：checkpoint称为检查点，是flink用于容错的机制，checkpoint由JM中的checkpoint协调器进行周期性的触发，
         * 然后Source端进行备份，然后生成barrier并且向下游传递，下游算子收到上游传来的barrier之后做自己状态的备份，当所有的完成备份之后
         * 整个checkpoint就完成了。如果当程序发生故障，可以从上一个检查点启动，程序会从上一个检查点继续处理，就像没有发生过故障一样。
         *
         * checkpoint的分类
         * 1、barrier对其：下游算子，只有收到上游所有相同编号的barrier之后才会进行checkpoint
         *      1.1 精确一次性
         *          先到的barrier会阻塞后面的数据
         *      1.2 至少一次性
         *          先到的barrier不会阻塞后面的数据
         * 2、非对齐barrier：先到的barrier也是触发checkpoint
         *     设置非对齐检查点，checkpoint的语义级别一定要设置为精确一次性。
         *     精确一次性的保证方式：先到的barrier会越过缓冲区的数据，直接到达算子的输出缓冲区，然后将当前算子的输入和输出缓冲器区的数据存储起来
         *     同时将还未到的barrier之后的数据也都缓存起来。通过这种方式来保证精准一次性。但是这样显然增加了数据的存储量，同时增加了IO的压力
         * 3、通用增量 checkpoint (changelog)
         *    在flink1.15之前只有rocksDB可以实现增量检查点，但是changelog可以使的HashMap也实现增量检查dian
         *    实现的原理是：
         *      flink会为带状态的算子维护一个changelog和state table
         *          changelog中存储的是状态变更的日志，这个日志会持久化的远端。
         *          state table存储的是最终的状态。间隔一段的时间就会将最终的状态写入到远端存储系统，然后将最终状态之前的变更日志都删除掉
         *
         *      使用时的注意事项
         *          （1）目前标记为实验性功能，开启后可能会造成资源消耗增大：
         *              	HDFS上保存的文件数变多
         *              	消耗更多的IO带宽用于上传变更日志
         *              	更多的CPU用于序列化状态更改
         *              	TaskManager使用更多内存来缓存状态更改
         *          （2）使用限制：
         *              	Checkpoint的最大并发必须为1
         *              	从 Flink 1.15 开始，只有文件系统的存储类型实现可用（memory测试阶段）
         *              	不支持 NO_CLAIM 模式
         *  3、最终检查点
         *      在批处理程序中，因为数据是有边界的，如果发生数据倾斜，导致有些子任务的数据很快就处理完了，有些子任务还在处理中
         *      如果发送checkpoint,那些已经关闭的子任务是否参与到checkpoint中。通常我们是开启的
         */
    }
}
